package cn.doitedu.cn.doitedu.rtdw.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/2
 * @Desc: 学大数据，上多易教育
 *   订单日清日结看板
 **/
public class Job5_OrderDaySettlement {

    public static void main(String[] args) {
        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        // 添加开启周期行提前触发的 窗口触发器
        TableConfig config = tenv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled",true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay","1000 ms");


        // 建表，映射 业务库中  订单表： oms_order
        tenv.executeSql(
                "CREATE TABLE order_mysql (                         " +
                        "      id BIGINT,                                    " +  // 订单id
                        "      status INT,                                   " +  // 订单状态
                        "      total_amount decimal(10,2),                   " +  // 原价总额
                        "      pay_amount decimal(10,2),                     " +  // 应付(实付)金额
                        "      create_time timestamp(3),                     " +  // 创建时间
                        "      payment_time timestamp(3),                    " +  // 支付时间
                        "      delivery_time timestamp(3),                   " +  // 发货时间
                        "      receive_time timestamp(3),                    " +   //  确认收货时间
                        "      modify_time timestamp(3),                     " +  // 数据更新时间
                        "      rt as modify_time,                            " +  // 数据更新时间
                        "      watermark for rt as rt - interval '0' second, " +  // 数据更新时间
                        "     PRIMARY KEY (id) NOT ENFORCED       " +
                        "     ) WITH (                            " +
                        "     'connector' = 'mysql-cdc',          " +
                        "     'hostname' = 'doitedu'   ,          " +
                        "     'port' = '3306'          ,          " +
                        "     'username' = 'root'      ,          " +
                        "     'password' = 'root'      ,          " +
                        "     'database-name' = 'realtimedw',     " +
                        "     'table-name' = 'oms_order'          " +
                        ")"
        );

        /*   每秒更新：
             订单总数、总额 、应付总额 （当日新订单，累计到此刻,）
             待支付订单数、订单额  （当日新订单，累计到此刻）
             已支付订单数、订单额  （当日支付，累计到此刻）
             已发货订单数、订单额  （当日发货，累计到此刻）
             已完成订单数、订单额  （当日确认，累计到此刻）
         */
        tenv.executeSql("SELECT\n" +
                "   count(1) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE )  as  `订单总数`,\n" +
                "   sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE )  as  `订单原价总额`,\n" +
                "   sum(pay_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE )  as  `订单应付总额`,\n" +
                "   count(1) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE and status = 0)  as  `待支付总数`,\n" +
                "   sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd') = CURRENT_DATE and status = 0)  as  `待支付总额`\n" +
                "FROM order_mysql\n" +
                "GROUP BY \n" +
                "    TUMBLE(rt,INTERVAL '24' HOUR)\n").print();








    }

}
